package com.ruoyi.system.service.impl;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.alibaba.fastjson2.JSON;
import com.ruoyi.common.utils.SecurityUtils;
import com.ruoyi.system.domain.SysTalkBody;
import com.ruoyi.system.domain.SysTalkMessage;
import com.ruoyi.system.domain.resp.SysTalkBodyResp;
import com.ruoyi.system.listener.CompletedCallBack;
import com.ruoyi.system.listener.SparkAISubscriber;
import com.ruoyi.system.service.ISparkApiService;
import com.ruoyi.system.uitls.UserSessionUtil;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.security.web.FilterChainProxy;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;
import java.util.List;

/**
 * 星火服务实现
 */
@Slf4j
@Service
public class SparkApiImpl implements ISparkApiService, CompletedCallBack {

    @Value("${sparkApi.url}")
    private String sparkApiUrl;

    @Value("${sparkApi.appId}")
    private String appId;

    @Value("${sparkApi.apiSecret}")
    private String apiSecret;

    @Value("${sparkApi.apiKey}")
    private String apiKey;

    private WebClient webClient;
    @Resource
    private UserSessionUtil userSessionUtil;


    public SparkApiImpl() {
        log.info("initProd");
        this.webClient = WebClient.builder()
                .defaultHeader(HttpHeaders.CONTENT_TYPE, "application/json")
                .build();

    }

    @Override
    public SysTalkBodyResp communication(SysTalkBody sysTalkBody) {
        HttpResponse response = HttpRequest.post(sparkApiUrl)
                .header("Content-Type", "application/json")
                .header("Authorization", "Bearer " + apiKey + ":" + apiSecret)
                .body(JSON.toJSONString(sysTalkBody))
                .execute();
        String body = response.body();
        int status = response.getStatus();
        if (status == 200) {
            log.info("请求成功，响应内容：" + body);
        } else {
            log.info("请求失败，状态码：" + status + "，响应内容：" + body);
        }
        return JSON.parseObject(body, SysTalkBodyResp.class);
    }

    @Override
    public Flux<String> communicationStream(SysTalkBody sysTalkBody) {
        List<SysTalkMessage> messages = sysTalkBody.getMessages();
        if (!messages.isEmpty()) {
            SysTalkMessage sysTalkMessage = messages.get(messages.size() - 1);
            log.info("prompt:{}", sysTalkMessage.getContent());
        } else {
            log.info("prompt:{}", "无对话");
        }

        return Flux.create(emitter -> {
            //订阅者订阅
            SparkAISubscriber subscriber = new SparkAISubscriber(emitter, this);
            Flux<String> SparkAiResponse =
                    getSparkApiStream(sysTalkBody);

            SparkAiResponse.subscribe(subscriber);
            emitter.onDispose(subscriber);
        });
        //将存入对话历史
//        userSessionUtil.addMessage();
    }

    private Flux<String> getSparkApiStream(SysTalkBody sysTalkBody) {
        return webClient.post()
                .uri(sparkApiUrl)
                .header(HttpHeaders.AUTHORIZATION, "Bearer " + apiKey + ":" + apiSecret)
                .bodyValue(JSON.toJSONString(sysTalkBody))
                .retrieve()
                .bodyToFlux(String.class)
                .doOnNext(data -> log.info("SparkAI返回数据：{}", data))
                .onErrorResume(WebClientResponseException.class, ex -> {
                    HttpStatus status = ex.getStatusCode();
                    String res = ex.getResponseBodyAsString();
                    log.error("Spark API error: {} {}", status, res);
                    return Mono.error(new RuntimeException(res));
                })
                .doOnComplete(() -> log.info("SparkAI返回数据结束了"));
    }

    @Override
    public void completed(SysTalkMessage questions, String response) {
        log.info("questions:{},response:{}", questions.getContent(), response);
    }

    @Override
    public void fail(SysTalkMessage questions, String response) {
        log.error("questions:{},response:{}", questions.getContent(), response);
    }

    @Override
    public void clearHistory(String sessionId) {
        //暂未实现
    }
}
